"""
本层为数据提供服务层,在数据库基础操作层上,设计针对用户的接口,原始数据的清洗筛选等操作也在本层完成
"""
from collections import defaultdict
from typing import Tuple,Optional
import numpy as np
from sqlalchemy import create_engine
from sqlalchemy.dialects.mysql import insert as mysql_insert  # MySQL方言支持
from .SqlBaseOperation import *
from .tables import *
from gm.api import history,set_token
from datetime import datetime,timedelta
import json 
from .utils import convert_time
import os
from dotenv import load_dotenv
from abs.DataServiceInterface import IDatabaseManager,IDataService

class SQLServiceMgr(IDataService,IDatabaseManager):
    
    def __init__(self):
        load_dotenv('env.env')  # 加载.env文件
        db_user = os.getenv('DB_USER', 'root')
        db_pass = os.getenv('DB_PASS', 'default_pass')
        db_host = os.getenv('DB_HOST', 'localhost')
        db_name = os.getenv('DB_NAME', 'quantdata')
        self.engine = create_engine(f'mysql+mysqlconnector://{db_user}:{db_pass}@{db_host}:3306/{db_name}')
        set_token(os.getenv('GM_TOKEN'))
        self.table_mapper = {'symbol_min':symbol_min,'symbol_tick':symbol_tick,'name':symbol_name,'group':symbol_group}

    def _get_conn(self):
        """
            创建一个新的数据库连接
        """
        return self.engine.connect()

    def history(self,symbols:list, start_time:datetime|str, end_time:datetime|str, frequency:str,
                        fields=None, filter_st=False)->pd.DataFrame:
        """
            获取标的的历史交易数据,返回dataframe
            Args:
                frequency: tick,min,day,week,month,year
        """
        with self._get_conn() as conn:
            start_time = convert_time(start_time)
            end_time = convert_time(end_time)                  
            return get_data(conn,symbols,start_time,end_time,frequency,fields,filter_st)
    
    def change_symbol_name(self,symbol:str,nname:str):
        """
            修改某个标的/etf/指数的名称
        """
        with self._get_conn() as conn:
            with conn.begin():
                update_data(conn,symbol_name,symbol_name.c.symbol==symbol,{'symbol_name':nname})

    def _update_time_range_on_delete(self, 
        conn,
        symbol: str, 
        delete_start: datetime, 
        delete_end: datetime, 
        data_type: str
    ):
        """根据删除时间范围调整时间跨度表"""
        # 确定目标表
        range_table = symbol_min_range if data_type == 'min' else symbol_tick_range
        
        condition = (range_table.c.symbol == symbol) & \
                   (range_table.c.end_time >= delete_start) & \
                   (range_table.c.start_time <= delete_end)
        
        # 查询现有时间段
        query = select(range_table).where(condition)
        existing = pd.read_sql(query, conn)

        # 增加时区
        existing['start_time'] = pd.to_datetime(existing['start_time']).dt.tz_localize('Asia/Shanghai')
        existing['end_time'] = pd.to_datetime(existing['end_time']).dt.tz_localize('Asia/Shanghai')
        
        # 遍历可能受影响的时间段
        new_ranges = []
        delete_conditions = []
        
        for _, row in existing.iterrows():
            original_start = row['start_time']
            original_end = row['end_time']
            
            # 收集删除条件
            delete_conditions.append(
                (range_table.c.symbol == symbol) & 
                (range_table.c.start_time == original_start)
            )
               
            # Case 1: 原始区间在删除范围左侧未覆盖部分
            if original_start < delete_start:
                new_end = delete_start - timedelta(microseconds=1)
                if new_end > original_start:
                    new_ranges.append({
                        'symbol': symbol,
                        'start_time': original_start,
                        'end_time': new_end
                    })
            
            # Case 2: 原始区间在删除范围右侧未覆盖部分
            if original_end > delete_end:
                new_start = delete_end + timedelta(seconds=1)
                if new_start < original_end:
                    new_ranges.append({
                        'symbol': symbol,
                        'start_time': new_start,
                        'end_time': original_end
                    })
        
        # 删除旧记录
        if delete_conditions:
            conn.execute(
                range_table.delete().where(or_(*delete_conditions))
            )

        # 插入新生成的有效区间
        if new_ranges:
            conn.execute(range_table.insert(), new_ranges)


    def delete_symbol(self, symbol: str, table_type: str, start_time: datetime | str = None, end_time: datetime | str = None,force:bool=False):
        """删除某个标的历史数据（精确维护时间跨度表）"""
        if(not force):
            print('cancel delete')
            return 
        with self._get_conn() as conn:
            if table_type not in ('min', 'tick'):
                raise ValueError(f"无效数据类型: {table_type}，仅支持'min'或'tick'")

            with conn.begin() as transaction:
                try:
                    # 转换时间参数
                    del_start = convert_time(start_time) if start_time else None
                    del_end = convert_time(end_time) if end_time else None

                    # 删除主数据表
                    if table_type == 'min':
                        # 删除分钟数据
                        condition = symbol_min.c.symbol == symbol
                        if del_start and del_end:
                            condition &= symbol_min.c.time.between(del_start, del_end)
                        delete_data(conn, symbol_min, condition)

                        # 更新分钟时间跨度表
                        if del_start and del_end:
                            self._update_time_range_on_delete(conn,symbol, del_start, del_end, 'min')
                        else:  # 删除全部时直接清理
                            conn.execute(symbol_min_range.delete().where(symbol_min_range.c.symbol == symbol))

                    elif table_type == 'tick':
                        condition = symbol_tick.c.symbol == symbol
                        if del_start and del_end:
                            condition &= symbol_tick.c.time.between(del_start, del_end)
                        delete_data(conn, symbol_tick, condition)

                        # 更新分钟时间跨度表
                        if del_start and del_end:
                            self._update_time_range_on_delete(conn, symbol, del_start, del_end, 'tick')
                        else:  # 删除全部时直接清理
                            conn.execute(symbol_tick_range.delete().where(symbol_tick_range.c.symbol == symbol))

                    # 如果标的在 symbol_group 表中，删除 symbol_name 表记录
                    if check(conn, symbol_group, symbol):
                        delete_data(conn, symbol_name, symbol_name.c.symbol == symbol)

                except Exception as e:
                    transaction.rollback()
                    raise RuntimeError(f"删除失败: {str(e)}")

    def delete_group(self,symbol:str):
        """
            删除某个标的集合的包含标的信息
        """
        with self._get_conn() as conn:
            with conn.begin():
                condition = symbol_group.c.symbol==symbol
                delete_data(conn,symbol_group,condition)
                if not check(conn, symbol_min, symbol) and not (check(conn, symbol_tick, symbol)):
                    delete_data(conn, symbol_name, symbol_name.c.symbol == symbol)
    
    def _update_time_ranges(self, conn, symbol_ranges: list, data_type: str):
        """
        批量更新多个标的的时间范围（使用symbol+start_time作为唯一标识）
        :param conn: 数据库连接
        :param symbol_ranges: 列表，每个元素为元组(symbol, new_start, new_end)
        :param data_type: 'min' 或 'tick'
        """
        if not symbol_ranges:
            return
        
        range_table = symbol_min_range if data_type == 'min' else symbol_tick_range
        
        # 1. 获取所有相关标的的现有时间范围
        symbols = [sr[0] for sr in symbol_ranges]
        existing_query = select(range_table).where(range_table.c.symbol.in_(symbols))
        existing_rows = conn.execute(existing_query).fetchall()
        
        # 2. 构建现有范围的映射 {symbol: [(symbol, start, end)]}
        existing_map = defaultdict(list)
        for row in existing_rows:
            existing_map[row.symbol].append((row.symbol, row.start_time, row.end_time))
        
        # 3. 准备新旧范围合并
        to_delete = []  # 待删除的旧记录条件列表
        to_insert = []  # 待插入的新记录
        
        for symbol, new_start, new_end in symbol_ranges:
            merged_start = new_start
            merged_end = new_end
            
            # 收集要删除的旧记录
            for sym, estart, eend in existing_map.get(symbol, []):
                if (estart <= new_end + timedelta(seconds=1)) and (eend >= new_start - timedelta(seconds=1)):
                    to_delete.append((sym, estart))  # 使用symbol+start_time作为删除条件
                    merged_start = min(merged_start, estart)
                    merged_end = max(merged_end, eend)
            
            # 添加合并后的范围
            to_insert.append({
                'symbol': symbol,
                'start_time': merged_start,
                'end_time': merged_end
            })
        
        # 4. 执行批量操作
        if to_delete:
            # 构建复合删除条件
            delete_conditions = or_(
                *[and_(range_table.c.symbol == sym, 
                    range_table.c.start_time == start)
                for sym, start in to_delete]
            )
            conn.execute(range_table.delete().where(delete_conditions))
        
        if to_insert:
            conn.execute(range_table.insert(), to_insert)

    def insert_new_symbol(self, data: pd.DataFrame, data_type: str):
        """插入多标的数据（分钟级或tick级），自动按标的去重"""
        type_rules = {
            'min': {
                'required': ['symbol', 'time', 'open', 'close', 'high', 'low', 'volume', 'amount'],
                'time_col': 'time',
                'table': symbol_min
            },
            'tick': {
                'required': ['symbol', 'price', 'created_at','open','high','low','cum_volume','cum_amount','trade_type','last_volume','cum_position','last_amount','iopv','quotes'],
                'time_col': 'created_at',
                'table': symbol_tick
            }
        }

        if data_type not in type_rules:
            raise ValueError(f"无效数据类型: {data_type}，仅支持'min'或'tick'")
        rule = type_rules[data_type]

        missing_cols = [col for col in rule['required'] if col not in data.columns]
        if missing_cols:
            raise ValueError(f"{data_type}数据缺失必要字段: {missing_cols}")

        # 按标的分组处理
        grouped = data.groupby('symbol')
        symbol_ranges = []  # 收集所有标的的时间范围
        
        for symbol, group in grouped:
            time_col = rule['time_col']
            min_time = group[time_col].min()
            max_time = group[time_col].max()
            symbol_ranges.append((symbol, min_time, max_time))
        with self._get_conn() as conn:
            with conn.begin() as transaction:
                try:
                    # 批量更新时间范围
                    self._update_time_ranges(conn, symbol_ranges, data_type)
                    # 批量插入所有数据（使用ON DUPLICATE KEY UPDATE）
                    stmt = mysql_insert(rule['table']).values(data.to_dict(orient='records'))
                    update_stmt = stmt.on_duplicate_key_update(
                        {col: stmt.inserted[col] for col in data.columns}
                    )
                    conn.execute(update_stmt)
                    
                except Exception as e:
                    transaction.rollback()
                    raise RuntimeError(f"{data_type}数据插入失败: {str(e)}")

    def insert_new_group(self, symbol: str, position: list = None):
        """
        添加新的标的集合（存在则更新position）
        """
        # 2. 将position列表序列化为JSON字符串
        group_data = {
            "symbol": symbol,
            "position": json.dumps(position) if position else json.dumps([])  # 强制转换为JSON字符串
        }
        
        # 使用MySQL方言的insert语句
        stmt = mysql_insert(symbol_group).values(group_data)  # 表名修正为group
        
        # 定义更新操作（需同步序列化）
        update_stmt = stmt.on_duplicate_key_update(
            position=json.dumps(position) if position else json.dumps([])
        )
        
        # 执行并提交
        with self._get_conn() as conn:
            with conn.begin():
                conn.execute(update_stmt)

    def insert_new_name(self, symbol: str, name_str: str):
        """
        插入或更新标的名称（存在则更新名称）
        """
        stmt = mysql_insert(symbol_name).values(symbol=symbol, symbol_name=name_str)
        update_stmt = stmt.on_duplicate_key_update(symbol_name=name_str)
        with self._get_conn() as conn:
            with conn.begin():
                conn.execute(update_stmt)  

    def get_df(self,symbol:list|str,
            start_time:str|datetime,
            end_time:str|datetime,
            frequency:str,
            field:list = None)->pd.DataFrame:
        """
            通过掘金api获取dataframe格式的交易数据
        """
        return history(symbol=symbol,frequency=frequency,start_time=start_time,end_time=end_time,fields=field,df=True)

    def show_exist_symbols(self, type: str):
        """获取当前存储的所有symbol以及时间范围"""
        with self._get_conn() as conn:
            with conn.begin():
                symbols = get_symbols(conn, type)
                # 确保始终返回有明确列名的DataFrame，即使数据为空
                data = []
                for symbol in symbols:
                    ranges = get_symbol_ranges(conn, symbol, type)
                    for start_time, end_time in ranges:
                        data.append({'symbol': symbol, 'time_range': (start_time, end_time)})
        
        # 显式指定列名创建DataFrame
        df = pd.DataFrame(data, columns=['symbol', 'time_range'])
        return df

    def update_today_data(self,min:bool=True,tick:bool=False):
        """
            更新今日的交易数据(只会更新表中已经存在的标的数据)
        """

        now = datetime.now()
        if (now.hour<15):
            print('更新时间太早,请在收盘后(15:00)更新')
            return
        pre = now.replace(hour=0, minute=0, second=0, microsecond=0)
        fields_dict = {
                        'min':['symbol','open','close','high','low','amount','volume','bob'],
                       'tick':['symbol', 'price', 'created_at','open','high','low','cum_volume','cum_amount','trade_type','last_volume','cum_position','last_amount','iopv','quotes']
                       }
        def _insert(conn,type:str):
            symbols = get_symbols(conn,type)
            freq = '60s' if type == 'min' else 'tick'
            data = self.get_df(symbols,pre,now,freq,fields_dict[type])
            if(data.shape[0]==0):
                print('今日未开盘!')
                return
            if type=='min':
                data.rename(columns={'bob':'time'},inplace=True)
            self.insert_new_symbol(data,type)

        with self._get_conn() as conn:
            if(min):
                _insert(conn,'min') 
            if(tick):
                _insert(conn,'tick')
    
    def get_group_positions(self, symbols: list) -> list:
        """
        查询标的集合的所有标的编号，返回列表的列表（外层列表按输入symbols顺序，内层为标的代码列表）
        """
        if(not isinstance(symbols,list)):
            raise ValueError("symbols应该是list格式")
        with self._get_conn() as conn:
            with conn.begin():
                res = conn.execute(
                    select(symbol_group.c.position).where(symbol_group.c.symbol.in_(symbols))
                )
                positions = []
                for row in res:
                    try:
                        # 解析JSON字符串为列表
                        parsed = json.loads(row[0]) if row[0] else []
                        # 确保元素为字符串类型（防御性处理）
                        parsed = [str(item) for item in parsed]
                        positions.append(parsed)
                    except (json.JSONDecodeError, TypeError):
                        # 记录错误或返回空列表
                        positions.append([])
                return positions
    
    def get_backtest_list(self) -> list:
        """获取所有回测的名称列表"""
        with self._get_conn() as conn:
            with conn.begin():
                return get_backtest_list(conn)
    
    def delete_backtest(self, backtest_name: str):
        """删除回测结果"""
        with self._get_conn() as conn:
            with conn.begin():
                delete_backtest(conn,backtest_name)

    def save_backtest_result(self, summary: dict, daily_data: pd.DataFrame, trade_details: pd.DataFrame = None):
        """
        保存回测结果（包含交易细节）
        :param summary: 摘要信息字典
        :param daily_data: 每日资产变化DataFrame
        :param trade_details: 交易细节DataFrame（可选）
        """
        # 准备每日数据
        if trade_details is not None:
            # 将NaN替换为None，避免插入数据库时出错
            trade_details = trade_details.replace({np.nan: None})
            
        daily_records = []
        for date, row in daily_data.iterrows():
            daily_records.append({
                'backtest_name': summary['name'],
                'date': date,
                'cash': row['cash'],
                'position_value': row.get('position', 0),
                'total_asset': row['total'],
                'daily_return': row.get('daily_return', 0)  
            })
        
        # 准备交易记录
        trade_records = []
        if trade_details is not None:
            for idx, row in trade_details.iterrows():
                trade_records.append({
                    'backtest_name': summary['name'],
                    'trade_id': idx,  # 使用索引作为trade_id
                    'date': row['date'],
                    'symbol': row['symbol'],
                    'action': row['action'],
                    'price': row['price'],
                    'amount': row['amount'],
                    'fee': row['fee'],
                    'value': row.get('value', None),
                    'profit': row.get('profit', None),
                    'return_pct': row.get('return_pct', None)
                })
        
        with self._get_conn() as conn:
            with conn.begin():
                # 保存摘要
                stmt = mysql_insert(backtest_summary).values(summary)
                update_stmt = stmt.on_duplicate_key_update(
                    {k: stmt.inserted[k] for k in summary.keys()}
                )
                conn.execute(update_stmt)
                
                # 保存每日数据
                conn.execute(
                    backtest_daily.delete().where(backtest_daily.c.backtest_name == summary['name'])
                )
                if daily_records:
                    conn.execute(backtest_daily.insert(), daily_records)
                
                # 保存交易细节
                if trade_records:
                    conn.execute(
                        backtest_trades.delete().where(backtest_trades.c.backtest_name == summary['name'])
                    )
                    conn.execute(backtest_trades.insert(), trade_records)

    def get_backtest_detail(self, backtest_name: str) -> Tuple[dict, pd.DataFrame, Optional[pd.DataFrame]]:
        """获取回测的摘要、每日资产变化和交易细节"""
        with self._get_conn() as conn:
            # 获取摘要
            stmt = select(backtest_summary).where(backtest_summary.c.name == backtest_name)
            res = conn.execute(stmt)
            row = res.fetchone()
            if row is None:
                return None, None, None
            summary = {
                "name": row.name,
                "start_time": row.start_time,
                "end_time": row.end_time,
                "initial_cash": row.initial_cash,
                "final_value": row.final_value,
                "total_return": row.total_return,
                "annual_return": row.annual_return,
                "max_drawdown": row.max_drawdown,
                "total_fees": row.total_fees,
                "trade_days": row.trade_days,
                "total_trades": row.total_trades,
                "sharpe_ratio": row.sharpe_ratio,
                "sortino_ratio": row.sortino_ratio,
                "win_trades": row.win_trades,
                "loss_trades": row.loss_trades,
                "win_rate": row.win_rate,
                "avg_win": row.avg_win,
                "avg_loss": row.avg_loss,
                "profit_factor": row.profit_factor,
                "volatility": row.volatility,
                "max_drawdown_start": row.max_drawdown_start,
                "max_drawdown_end": row.max_drawdown_end,
                "max_drawdown_duration": row.max_drawdown_duration,
                "end_position_count": row.end_position_count,
                "end_position_ratio": row.end_position_ratio
            }
            
            # 获取每日数据
            stmt_daily = select(
                backtest_daily.c.date,
                backtest_daily.c.cash,
                backtest_daily.c.position_value.label('position'),
                backtest_daily.c.total_asset.label('total'),
                backtest_daily.c.daily_return
            ).where(backtest_daily.c.backtest_name == backtest_name)
            
            daily_df = pd.read_sql(stmt_daily, conn, index_col='date')
            daily_df.sort_index(inplace=True)
            
            # 获取交易细节
            stmt_trades = select(
                backtest_trades.c.date,
                backtest_trades.c.symbol,
                backtest_trades.c.action,
                backtest_trades.c.price,
                backtest_trades.c.amount,
                backtest_trades.c.fee,
                backtest_trades.c.value,
                backtest_trades.c.profit,
                backtest_trades.c.return_pct
            ).where(backtest_trades.c.backtest_name == backtest_name)
            
            trade_details_df = pd.read_sql(stmt_trades, conn)
            if not trade_details_df.empty:
                # 添加一个虚拟索引列用于排序
                trade_details_df = trade_details_df.sort_values('date').reset_index(drop=True)
            
            return summary, daily_df, trade_details_df
        
    